package com.tee.live.id.generate.service.impl;

import com.tee.live.id.generate.dao.mapper.IdGenerateMapper;
import com.tee.live.id.generate.dao.po.IdGeneratePO;
import com.tee.live.id.generate.rpc.IdGenerateRpcImpl;
import com.tee.live.id.generate.service.IdGenerateService;
import com.tee.live.id.generate.service.bo.LocalSeqIdBO;
import com.tee.live.id.generate.service.bo.LocalUnSeqIdBO;
import jakarta.annotation.Resource;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private final static Logger LOGGER = LoggerFactory.getLogger(IdGenerateRpcImpl.class);
    private static Map<Integer, LocalSeqIdBO> localSeqIdBOMap = new ConcurrentHashMap<>();
    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdBOMap = new ConcurrentHashMap<>();
    private static final int SEQ_TYPE = 1;
    private static ThreadPoolExecutor threadPoolExecutor =
        new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS,
                               new ArrayBlockingQueue<>(100), new ThreadFactory() {
            @Override
            public Thread newThread(final Runnable r) {

                Thread thread = new Thread(r);
                thread.setName("id-generate-thread-" + ThreadLocalRandom.current()
                    .nextInt(1000));
                return thread;
            }
        });
    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();

    private static final float UPDATE_RATE = 0.75f;
    @Resource
    private IdGenerateMapper idGenerateMapper;

    @Override
    public Long getSeqId(final Integer id) {

        if (id == null) {
            LOGGER.error("[getSeqId] id is null");
            return null;
        }
        LocalSeqIdBO localSeqIdBO = localSeqIdBOMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[getSeqId] localSeqIdBO is null");
            return null;
        }
        if (localSeqIdBO.getCurrentNum()
                .get() > localSeqIdBO.getNextThreshold()) {
            LOGGER.error("[getSeqId] id id over limit,id is {}", id);
            return null;
        }
        long returnId = localSeqIdBO.getCurrentNum()
            .getAndIncrement();
        if (returnId > localSeqIdBO.getNextThreshold()) {
            //同步去刷新
            LOGGER.error("[getSeqId] id is over limit,id is {}", id);
            return null;
        }
        return returnId;
    }

    @Override
    public Long getUnSeqId(final Integer id) {

        if (id == null) {
            LOGGER.error("[getUnSeqId] id is null");
            return null;
        }
        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdBOMap.get(id);
        if (localUnSeqIdBO == null) {
            LOGGER.error("[getUnSeqId] localUnSeqIdBO is null");
            return null;
        }
        Long returnId = localUnSeqIdBO.getIdQueue()
            .poll();
//        LOGGER.info("[getUnSeqId] return id {}", returnId);
        this.refreshUnLocalSeqId(localUnSeqIdBO);
        return returnId;
    }

    /**
     * 刷新本地有序id段
     *
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {

        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if (localSeqIdBO.getCurrentNum()
                .get() - localSeqIdBO.getCurrentStart() > step * UPDATE_RATE) {
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("[refreshLocalSeqId] semaphore is null,id is {}", localSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                // 异步进行id段操作
                threadPoolExecutor.execute(new Runnable() {
                    @Override
                    public void run() {

                        IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                        tryUpdateMySQLRecord(idGeneratePO);
                    }
                });
            }
            semaphore.release();

        }
    }

    /**
     * 刷新本地无序id段
     *
     * @param localUnSeqIdBO
     */
    private void refreshUnLocalSeqId(final LocalUnSeqIdBO localUnSeqIdBO) {

        Long start = localUnSeqIdBO.getCurrentStart();
        Long end = localUnSeqIdBO.getNextThreshold();
        if (localUnSeqIdBO.getIdQueue()
                .size() < (end - start) * (1 - UPDATE_RATE)) {
            // 刷新操作
            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("semaphore is null,id is {}", localUnSeqIdBO.getId());
                return;
            }
            boolean acquireStatus = semaphore.tryAcquire();
            if (acquireStatus) {
                threadPoolExecutor.execute(new Runnable() {
                    @Override public void run() {
                        try {
                            IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                            tryUpdateMySQLRecord(idGeneratePO);
                            LOGGER.info("无序id段同步完成,id is {}", localUnSeqIdBO.getId());
                        } catch (Exception e) {
                            LOGGER.error("[refreshUnLocalSeqId] error is ", e);
                        } finally {
                            semaphore.release();
                        }
                    }
                });
            }

        }
    }

    // 初始化回调
    @Override
    public void afterPropertiesSet() throws Exception {

        List<IdGeneratePO> idGeneratePOS = idGenerateMapper.selectAll();
        for (IdGeneratePO idGeneratePO : idGeneratePOS) {
            LOGGER.info("服务启动,抢占新id段");
            tryUpdateMySQLRecord(idGeneratePO);
            semaphoreMap.put(idGeneratePO.getId(), new Semaphore(1));
        }

    }

    /**
     * 更新mysql里面的分布式id的配置信息，占用相应的id段
     * 同步执行，很多的网络IO，性能较慢
     *
     * @param idGeneratePO
     */
    private void tryUpdateMySQLRecord(IdGeneratePO idGeneratePO) {

        int updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
        if (updateResult > 0) {
            localIdBOHandler(idGeneratePO);
            return;
        }
        for (int i = 0; i < 3; i++) {
            idGeneratePO = idGenerateMapper.selectById(idGeneratePO.getId());
            updateResult = idGenerateMapper.updateNewIdCountAndVersion(idGeneratePO.getId(), idGeneratePO.getVersion());
            if (updateResult > 0) {
                localIdBOHandler(idGeneratePO);
                return;
            }
        }

        throw new RuntimeException("表id段占用失败，竞争过于激烈，id is " + idGeneratePO.getId());
    }

    private void localIdBOHandler(IdGeneratePO idGeneratePO) {

        long currentStart = idGeneratePO.getCurrentStart();
        long nextThreshold = idGeneratePO.getNextThreshold();
        long currentNum = currentStart;
        if (idGeneratePO.getIsSeq() == 1) {

            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong atomicLong = new AtomicLong(currentNum);
            localSeqIdBO.setId(idGeneratePO.getId());
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);
            localSeqIdBOMap.put(idGeneratePO.getId(), localSeqIdBO);
        } else {
            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGeneratePO.getId());
            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);
            List<Long> list = new ArrayList<>();
            long begin = currentStart;
            long end = nextThreshold;
            for (long i = begin; i < end; i++) {
                list.add(i);
            }
            Collections.shuffle(list);
            ConcurrentLinkedQueue<Long> queue = new ConcurrentLinkedQueue<>(list);
            localUnSeqIdBO.setIdQueue(queue);
            localUnSeqIdBOMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);
        }
    }

}
